package com.grwx.omc.kafka.basekafka.producer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;


/**
 * @Description: 消息服务的生产者
 * @Param:
 * @return:
 * @Author: ManJiang
 * @Date: 2020/5/8
 */
@Component
public class KafkaProducer {

    private static Logger logger = LoggerFactory.getLogger(KafkaProducer.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 发送消息 指明对应topic
     * @param msg
     * @param topic
     */
    public void sendMsg(String msg, String topic){
        ListenableFuture listenableFuture = kafkaTemplate.send(topic,msg);

        //消息发送回调函数
        SuccessCallback successCallback =  new SuccessCallback() {
            @Override
            public void onSuccess(Object o) {
                logger.info("消息发送成功!");
            }
        };

        FailureCallback failureCallback = new FailureCallback() {
            @Override
            public void onFailure(Throwable throwable) {
                logger.info("消息发送失败");
            }
        };

        listenableFuture.addCallback(successCallback,failureCallback);
    }
}
